package com.ls.rabbitmqService;

import com.alibaba.fastjson.JSONObject;
import com.ls.jpushMessageClient.JPushUtils;
import com.ls.rabbitmqService.utils.ConfigService;
import com.rabbitmq.client.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.lang.management.ManagementFactory;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * rabbitmq 消费者
 * 利用多线程消费，这是针对消息没有顺序判断的
 * 如果生成者的消息有顺序，建议用单线程，或者别的办法
 */
public class RabbitmqConsumer {

    public static final ExecutorService queueService = Executors.newFixedThreadPool(1);

    public static final ExecutorService consumerService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()/2);

    static Logger logger = LoggerFactory.getLogger(RabbitmqConsumer.class);
    static final String PLATFORM_ANDROID="android";
    static final String PLATFORM_IOS="ios";
    static final String PLATFORM_WINPHONE="winphone";
    static final String PLATFORM_ALL="all";

    static final String PUSHTYPE_NOTIFICATION="notification";
    static final String PUSHTYPE_MESSAGE="message";

    static final String AREAAPPKEY_INTERVAL="_";

    public static  JSONObject pushConfig=null;

    /**
     * 消息队列名称字符串，格式x.x.x 中间代表地区 后面是第几个队列
     * 队列默认都是持久化的，但是消息暂时不需要持久化，路由器已默认的direct直连的方式
     */
    public static String defaultQueueName="rabbitmq.test.test.queue";
    //已经监控的队列,可中途加入进来
    public static Set<String> queueListeners=new HashSet<String>();

    public static void consumer(String queueName){
        RabbitmqChannelPool.RabbitmqChannel rabbitmqChannel = RabbitmqChannelPool.getChannel();
        if(rabbitmqChannel==null || rabbitmqChannel.getChannel()==null){
            logger.warn("rabbitmqChannel==null || rabbitmqChannel.getChannel()==null");
        }else{
            Channel channel = rabbitmqChannel.getChannel();
            logger.info("[*] consumer waiting for message  threadName:{},tcpId:{},channelNum:{},channelId:{},currentConnectionNum:{},maxConnectionNum:{},maxTcpChannelNum:{}",
                    Thread.currentThread().getName(),rabbitmqChannel.getConnection().getId(),
                    rabbitmqChannel.getConnection().getNewChannelNum(),channel.hashCode()
            ,RabbitmqConnectionPool.currentConnectionNum,RabbitmqConnectionPool.maxConnectionNum,RabbitmqChannelPool.maxTcpChannelNum);
            //logger.info("[*] waiting for message To exit press threadName:"+Thread.currentThread().getName());
            //queue的持久化设置需要与这个队列本省设置的一致否则报错
            try {
                channel.queueDeclare(queueName,true,false,false,null);
                channel.basicQos(1);
                final Channel copyChannel=channel;
                Consumer consumer=new DefaultConsumer(copyChannel){
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                        copyChannel.basicAck(envelope.getDeliveryTag(),false);
                        final String message=new String(body,"utf-8");
                        logger.info(String.format("[X] received %s",message));
                        consumerService.execute(new Runnable() {
                            @Override
                            public void run() {
                                handlerMessage(message);
                            }
                        });
                    }
                };
                /**
                 * 消息确认机制改为手动确认  copyChannel.basicAck(envelope.getDeliveryTag(),false);
                 */
                channel.basicConsume(queueName,false,consumer);
            } catch (IOException e) {
                e.printStackTrace();
            }

        }
    }

    private static void handlerMessage(String message){
        /**
         * 目前发现并发数量达到500以上，比如1000条的时候采用一个jpushClient，后面有些消息就会收不到，
         * 这个问题可使用多个创建多个jpushClient，估计不会达到这个量
         * 还可以多开几个消费者进程，最好放在不同机器上.
         *
         * 发现多使用几个jpushClient 也不好使，消息肯定是发给jpush，但是jpush貌似给丢弃啦，在后台查询不到
         */
        JSONObject jsonObject = JSONObject.parseObject(message);
        /**
         * uids 代表批量
         * rabbitmq 收到的格式{'uid':1,'area':'hb','pushType':'','sendType':'one','platformType':'all','data':{'message':'msg','key':'value'}}
         * key:value 是map传递过来的值  platformType平台类型android,ios,all
         * sendType 的值为one单条  batch批量 all所有用户
         */
        String sendType = jsonObject.getString("sendType");
        String pushType = jsonObject.getString("pushType");
        String platformType = jsonObject.getString("platformType");
        String area = jsonObject.getString("area");
        String appKey = jsonObject.getString("appKey");
        if(area==null){
            area="null";
        }
        if(appKey==null){
            appKey="null";
        }
        JSONObject data = jsonObject.getJSONObject("data");
        String[] alias_arr=null;
        List<JPushUtils.AreaConfig> areaConfigs = JSONObject.parseArray(RabbitmqConsumer.pushConfig.getString("areaConfigs"), JPushUtils.AreaConfig.class);
        String areaAppKey="";
        Set<String> areaAppKeys=new HashSet<>();
        if("batch".equals(sendType)){
            List<Integer> uids=null;
            if(jsonObject.containsKey("uids")){
                uids=(List<Integer>)jsonObject.get("uids");
            }
            if(uids!=null && uids.size()>0){
                alias_arr=new String[uids.size()];
                StringBuilder stringBuilder=null;
                int index=0;
                for(Integer uid:uids){
                    stringBuilder=new StringBuilder();
                    stringBuilder.append(area);
                    stringBuilder.append(uid);
                    alias_arr[index++]=stringBuilder.toString();
                }
            }
            for(JPushUtils.AreaConfig areaConfig:areaConfigs){
                if(area.equals(areaConfig.getArea()) && appKey.equals(areaConfig.getAppKey())){
                    areaAppKey=AREAAPPKEY_INTERVAL+areaConfig.getArea()+AREAAPPKEY_INTERVAL+areaConfig.getAppKey();
                    areaAppKeys.add(areaAppKey);
                    break;
                }
            }
            if("".equals(areaAppKey)){ //area地区的appKey都发送一遍
                for(JPushUtils.AreaConfig areaConfig:areaConfigs){
                    if(area.equals(areaConfig.getArea())){
                        areaAppKey=AREAAPPKEY_INTERVAL+areaConfig.getArea()+AREAAPPKEY_INTERVAL+areaConfig.getAppKey();
                        areaAppKeys.add(areaAppKey);
                    }
                }
            }
        }else if("one".equals(sendType)){
            int uid = jsonObject.containsKey("uid")?jsonObject.getIntValue("uid"):-1;
            if(uid>0){
                alias_arr=new String[1];
                alias_arr[0]=area+uid;
            }
            for(JPushUtils.AreaConfig areaConfig:areaConfigs){
                if(area.equals(areaConfig.getArea()) && appKey.equals(areaConfig.getAppKey())){
                    areaAppKey=AREAAPPKEY_INTERVAL+areaConfig.getArea()+AREAAPPKEY_INTERVAL+areaConfig.getAppKey();
                    areaAppKeys.add(areaAppKey);
                    break;
                }
            }
            if("".equals(areaAppKey)){ //area地区的appKey都发送一遍
                for(JPushUtils.AreaConfig areaConfig:areaConfigs){
                    if(area.equals(areaConfig.getArea())){
                        areaAppKey=AREAAPPKEY_INTERVAL+areaConfig.getArea()+AREAAPPKEY_INTERVAL+areaConfig.getAppKey();
                        areaAppKeys.add(areaAppKey);
                    }
                }
            }
        }else if("all".equals(sendType)){ //广播
            //需要给这个地区的所有appKey都进行广播一次
            for(JPushUtils.AreaConfig areaConfig:areaConfigs){
                if(area.equals(areaConfig.getArea())){
                    areaAppKey=AREAAPPKEY_INTERVAL+areaConfig.getArea()+AREAAPPKEY_INTERVAL+areaConfig.getAppKey();
                    JPushUtils.broadcastPlatformNotificationWithAlias(areaAppKey,platformType,data.getString("message"));
                }
            }
        }
        if(alias_arr!=null && alias_arr.length>0){
            boolean flag=PLATFORM_ALL.equals(platformType) || PLATFORM_ANDROID.equals(platformType) || PLATFORM_IOS.equals(platformType) || PLATFORM_WINPHONE.equals(platformType) ;
            if(flag) {
                Map<String,String> parameters=new HashMap<>();
                for (Map.Entry<String, Object> stringObjectEntry : data.entrySet()) {
                    String key = stringObjectEntry.getKey();
                    if(!"message".equals(key)){
                        parameters.put(key,stringObjectEntry.getValue().toString());
                    }
                }

                for(String areaappkey:areaAppKeys){
                    if(PUSHTYPE_NOTIFICATION.equals(pushType)){
                        JPushUtils.sendPlatformNotificationWithAlias(areaappkey, platformType, data.getString("message"),parameters, alias_arr);
                    }else if(PUSHTYPE_MESSAGE.equals(pushType)){
                        JPushUtils.sendPlatformMessageWithAlias(areaappkey, platformType, data.getString("message"),parameters, alias_arr);
                    }
                }
            }
        }
    }

    private static void testConsumer(){
        int nThreads=8;

        for(int i=0;i<nThreads;i++){
            final int k=i;
            final DateFormat dateFormat=new SimpleDateFormat("YYYY-MM-dd HH:mm:ss");
            queueService.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        Thread.sleep(1);
                        //System.out.println("currentTime:"+dateFormat.format(new Date())+",name:"+Thread.currentThread().getName());
                        if(k==6) {
                            consumer(defaultQueueName);
                            consumer(defaultQueueName);
                        }
                        else
                            consumer(defaultQueueName);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }

                }
            });
        }
    }
    public static void shundownListener(){
        Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
            @Override
            public void run() {
                logger.info("====rabbitmq consumer shundown");
                //释放jpush的长链接
                JPushUtils.relepaseJpushClient();
                queueService.shutdown();
                consumerService.shutdown();
            }
        }));
    }
    public static void startup(){
        ConfigService.initConfig();
    }
    private static void addQueueListener(final String queueName){
        queueService.execute(new Runnable() {
            @Override
            public void run() {
                logger.info("consumerQueueName:"+queueName);
                consumer(queueName);
            }
        });
    }
    public static void loadQueue(JSONObject jsonObject){
        String listenerQueues = jsonObject.getString("listenerQueues");
        List<String> queues = JSONObject.parseArray(listenerQueues, String.class);
        if(queues!=null && queues.size()>0){
            for(String queueName:queues){
                if(!queueListeners.contains(queueName)) {
                    //不存在添加，同时放到线程池里,启动一个消费者，目前是一个消费者监听一个队列
                    queueListeners.add(queueName);
                    addQueueListener(queueName);
                }
            }
        }
    }
    public static void main(String[] args){
        String name = ManagementFactory.getRuntimeMXBean().getName();
        //查看进程pid
        String pid = name.split("@")[0];
        logger.info("====rabbitmq consumer startup pid:"+pid);
        startup();
        shundownListener();
    }

}
